package com.johnguo.beam_test.java;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.GroupByKey;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Sum;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;

//--runner=SparkRunner
//--runner=FlinkRunner
//--runner=ApexRunner
public class CombineDemo {

	static final List<String> LINES = Arrays.asList("To be, or not to be: that is the question: ",
			"Whether 'tis nobler in the mind to suffer ", "The slings and arrows of outrageous fortune, ",
			"Or to take arms against a sea of troubles, ");

	public static void main(String... args) {
		// flinkrun

		PipelineOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().create();
		System.out.println(options);
		Pipeline p = Pipeline.create(options);
		PCollection<String> reader = p.apply(Create.of(LINES)).setCoder(StringUtf8Coder.of());
		PCollection<Integer> collection = reader.apply(ParDo.of(new DoFn<String, Integer>() {
			private static final long serialVersionUID = 1L;

			@ProcessElement
			public void processElement(ProcessContext c) {
				for (String word : c.element().split(" ")) {
					if (!word.isEmpty()) {
						// System.out.println(word);
						c.output(1);
					}
				}
			}
		}));
		// collection2.apply(new AverageFn());
		// collection2.apply(Combine.globally(new AverageFn()))
		collection.apply(Combine.globally(Sum.ofIntegers()).withoutDefaults())
				.apply(ParDo.of(new DoFn<Integer, String>() {
					private static final long serialVersionUID = 1L;
					@ProcessElement
					public void processElement(ProcessContext c) {
						System.out.println(c.element().toString());
						c.output(c.element().toString());
					}
				}));
		// .apply(TextIO.write().to("/root/test/out/"));
		p.run().waitUntilFinish();
	}
}
